package com.net.xpay.stable.job;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.net.common.concurrent.ThreadUtil;
import com.net.xpay.common.enums.config.CommonSysConfigKey;
import com.net.xpay.common.helper.CommonSysConfigHelper;
import com.net.xpay.common.helper.NoticeHelper;
import com.net.xpay.core.constant.PayslipAtomicInteger;
import com.net.xpay.core.manager.safePay.SafePayBatchManager;
import com.net.xpay.core.service.safePay.SafePayService;
import com.net.shard.ShardUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
public class SafePayBatchUpdateJob implements Runnable {

    private long shardId;

    private SafePayBatchManager safePayBatchManager;

    private SafePayService safePayService;

    private CommonSysConfigHelper commonSysConfigHelper;

    private NoticeHelper noticeHelper;

    private boolean isNoticed;


    private final ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setNameFormat("SafePayBatchUpdateJob-%d").build();

    //Common Thread Pool
    private final ExecutorService pool = new ThreadPoolExecutor(16, 16,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(20000), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());


    public SafePayBatchUpdateJob(Long shardId, SafePayBatchManager safePayBatchManager,
                                 SafePayService safePayService, CommonSysConfigHelper commonSysConfigHelper, NoticeHelper noticeHelper) {
        this.shardId = shardId;
        this.safePayBatchManager = safePayBatchManager;
        this.safePayService = safePayService;
        this.commonSysConfigHelper = commonSysConfigHelper;
        this.noticeHelper = noticeHelper;
    }

    @Override
    public void run() {
        try {
            ShardUtil.setShardContext(shardId);
            while (true) {
                try {
//                    if (PayslipAtomicInteger.getNum() > 0) {
//                        LocalDateTime nowTime = LocalDateTime.now();
//                        DayOfWeek dayOfWeek = nowTime.getDayOfWeek();
//                        if (dayOfWeek == DayOfWeek.MONDAY
//                                || dayOfWeek == DayOfWeek.WEDNESDAY
//                                || dayOfWeek == DayOfWeek.FRIDAY) {
//
//                        }else {
//                            PayslipAtomicInteger.setNum(0);
//                        }
//                    }
                    doUpdateSafePayBatch();
                } catch (Throwable e) {
                    log.error("doUpdateSafePayBatch", e);
                }
                ThreadUtil.safeSleep(500, "");
            }
        } finally {
            ShardUtil.cleanShardContext();
        }
    }

    private void doUpdateSafePayBatch() {
        //检测下服务有没有被暂停
        checkServiceIfPaused();

        List<Long> batchIds = safePayBatchManager.listNeedTransBatch();
        if (CollectionUtils.isEmpty(batchIds)) {
            return;
        }
        List<Future<Void>> futures = new ArrayList<>();

        for (Long batchId : batchIds) {
            Future<Void> future = pool.submit(() -> {
                try {
                    //子线程需要重设一下分片id
                    ShardUtil.setShardContext(shardId);
                    //检测下服务有没有被暂停
                    checkServiceIfPaused();
                    safePayService.processTransSafePayBatch(batchId);
                } catch (Exception e) {
                    log.error("doUpdateSafePayBatch error|batchId={}", batchId, e);
                    throw e;
                } finally {
                    ShardUtil.cleanShardContext();
                }
                return null;
            });
            futures.add(future);
        }

        //同步一下
        for (Future<Void> f : futures) {
            try {
                f.get();
            } catch (Exception e) {
                log.error("doUpdateSafePayBatch get error", e);
            }
        }
    }

    private void checkServiceIfPaused() {
        while (true) {
            Long pause_stable_service = commonSysConfigHelper.getLong(CommonSysConfigKey.PAUSE_STABLE_SERVICE);
            if (pause_stable_service != null && pause_stable_service == 1L) {
                //服务被暂停了
                if (!isNoticed) {
                    noticeHelper.sendSystemInfoNotice("安全付后台转账服务已暂停(" + shardId + ")", "text");
                    isNoticed = true;
                }
                log.warn("checkServiceIfPaused...|shardId={}|threadName={}", shardId, Thread.currentThread().getName());
                ThreadUtil.safeSleep(1000, "");
            } else {
                if (isNoticed) {
                    log.warn("checkServiceIfPaused reStart|shardId={}|threadName={}", shardId, Thread.currentThread().getName());
                    noticeHelper.sendSystemInfoNotice("安全付后台转账服务已恢复(" + shardId + ")", "text");
                }
                break;
            }
        }
        isNoticed = false;
    }
}
